package com.zuoshengli.service.impl;

import com.zuoshengli.api.KafkaService;
import com.zuoshengli.api.OrderService;
import org.apache.dubbo.config.annotation.Service;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @Description TODO
 * @Author zuoshengli
 **/
@Service(version = "${demo.service.version}")
public class KafkaServiceImpl implements KafkaService {

    private static Logger logger = LoggerFactory.getLogger(KafkaTemplate.class);
    @Autowired
    private OrderService orderService;
    @Autowired
    private KafkaTemplate kafkaTemplate;

    @Override
    @KafkaListener(topics = "#{kafkaTopicName}", groupId = "#{topicGroupId}")
    public void processMessage(ConsumerRecord<String, String> record) throws Exception {
        logger.info("kafka processMessage start");
        logger.info("processMessage, topic = {}, msg = {}", record.topic(), record.value());

        orderService.createOptimisticOrderUseRedis(Integer.parseInt(record.value()));

        logger.info("kafka processMessage end");
    }

    @Override
    public void sendMessage(String topic, String data) {
        ListenableFuture<SendResult<String, String>> future =kafkaTemplate.send(topic, data);
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                logger.error("kafka send message error, ex = {}", throwable);
            }

            @Override
            public void onSuccess(SendResult<String, String> stringStringSendResult) {
                logger.info("kafka send message success, topic={}, data={}", topic, data);
            }
        });
    }
}
